package com.dw.common.service.impl;

import com.dw.common.configuration.Configuration;
import com.dw.common.mapper.KeyValueMapper;
import com.dw.common.service.KVService;
import io.etcd.jetcd.ByteSequence;
import io.etcd.jetcd.Client;
import io.etcd.jetcd.KV;
import io.etcd.jetcd.KeyValue;
import io.etcd.jetcd.Lease;
import io.etcd.jetcd.kv.GetResponse;
import io.etcd.jetcd.lease.LeaseKeepAliveResponse;
import io.etcd.jetcd.options.GetOption;
import io.etcd.jetcd.options.PutOption;
import io.grpc.stub.StreamObserver;
import org.springframework.stereotype.Service;

import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.concurrent.ExecutionException;

@Service
public class KVServiceImpl implements KVService {
    private final Client client;
    private final KV kvClient;
    private final Lease leaseClient;

    public KVServiceImpl() {
        this.client = ClientManager.getClent();
        this.kvClient = client.getKVClient();
        this.leaseClient = client.getLeaseClient();
    }

    @Override
    public void put(String key, String value) {
        try {
            kvClient.put(ByteSequence.from(key.getBytes(StandardCharsets.UTF_8)),
                    ByteSequence.from(value.getBytes(StandardCharsets.UTF_8))).get();
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
        }
    }

    @Override
    public String get(String key) {
        try {
            GetResponse response = kvClient.get(ByteSequence.from(key.getBytes(StandardCharsets.UTF_8))).get();
            if (response.getKvs().size() == 0) {
                return null;
            }
            return response.getKvs().get(0).getValue().toString(StandardCharsets.UTF_8);
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
        }
        return null;
    }

    @Override
    public List<String> getWithPrefix(String key) {
        try {
            GetOption option = GetOption.newBuilder().withPrefix(StringToBytes(key)).build();
            List<KeyValue> keyValues = kvClient.get(StringToBytes(key), option).get().getKvs();
            return KeyValueMapper.BytesToString(keyValues);
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
        }
        return null;
    }

    @Override
    public void putWithLease(String key, String value) {
        PutOption option = PutOption.newBuilder().withLeaseId(acquireActiveLease()).build();
        kvClient.put(StringToBytes(key), StringToBytes(value), option);
    }

    @Override
    public void putWithLease(String key, Object value) {
        putWithLease(key, value.toString());
    }

    private ByteSequence StringToBytes(String value) {
        return ByteSequence.from(value.getBytes(StandardCharsets.UTF_8));
    }

    public long acquireActiveLease() {
        try {
            long leaseId = leaseClient.grant(Configuration.LEASE_TTL).get().getID();
            leaseClient.keepAlive(leaseId, new StreamObserver<LeaseKeepAliveResponse>() {
                @Override
                public void onNext(LeaseKeepAliveResponse leaseKeepAliveResponse) {
                }

                @Override
                public void onError(Throwable throwable) {
                }

                @Override
                public void onCompleted() {
                }
            });
            return leaseId;
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
        }
        return 0;
    }

    @Override
    public void close() {
        this.kvClient.close();
        this.leaseClient.close();
        this.client.close();
    }

    static class ClientManager {
        private static final String ENDPOINT = "http://118.31.227.231:2379";
        private static final ByteSequence USER_NAME = ByteSequence.from("root".getBytes(StandardCharsets.UTF_8));
        private static final ByteSequence PASSWORD = ByteSequence.from("Dw981020".getBytes(StandardCharsets.UTF_8));

        private static Client client;

        public static Client getClent() {
            if (client == null) {
                synchronized (ClientManager.class) {
                    if (client == null) {
                        client = Client.builder()
                                .endpoints(ENDPOINT)
                                .user(USER_NAME)
                                .password(PASSWORD)
                                .build();
                    }
                }
            }
            return client;
        }
    }
}
